package com.bossien.flink.common.sqlcommand;

import com.bossien.common.utils.StringUtils;
import com.bossien.flink.annatation.BSelect;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.ibatis.reflection.ExceptionUtil;

import java.io.Serializable;
import java.lang.annotation.Annotation;
import java.lang.invoke.MethodHandles;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Parameter;

/**
 * @author 胡春林
 * mapper代理类
 */
@Slf4j
public class MapperProxy<T> implements InvocationHandler,Serializable {

    private  Class<T> iMapper;

    public MapperProxy(Class<T> iMapper) {
        this.iMapper =iMapper;
    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        Table result = null;
        try {
            if (Object.class.equals(method.getDeclaringClass())) {
                return method.invoke(this, args);
            }

            if (this.isDefaultMethod(method)) {
                return this.invokeDefaultMethod(proxy, method, args);
            }
        } catch (Throwable var5) {
            throw ExceptionUtil.unwrapThrowable(var5);
        }
        //获取method注解
        Annotation[] annotations = method.getDeclaredAnnotations();
        if(annotations == null || annotations.length == 0)
        {
            throw  new Exception("该mapper方法没有标明注解无法获取sql串，请确认");
        }
        for (Annotation annotation : annotations)
        {
            //开始进行判断
            if(!(annotation instanceof BSelect))
            {
             continue;
            }
            String executeSql = ((BSelect) annotation).value();
            log.info("当前执行方法为：{},执行sql语句为：{}",method.getName(),executeSql);
            Parameter[] parameters = method.getParameters();
            if(parameters == null || parameters[0] == null || StringUtils.isBlank(executeSql) || parameters.length != 1 )
            {
                throw new Exception("参数/SQL语句不正确，请检查(参数只能传入table类型)");
            }
            if(parameters[0].getParameterizedType().getClass().isInstance(Table.class))
            {
                Table table = (Table) args[0];
                //开始执行sql语句
                TableEnvironment tableEnvironment = table.tableEnv();
                result = tableEnvironment.sqlQuery(executeSql);
                return result;
            }
        }

        return result;

    }
    private Object invokeDefaultMethod(Object proxy, Method method, Object[] args) throws Throwable {
        Constructor<MethodHandles.Lookup> constructor = MethodHandles.Lookup.class.getDeclaredConstructor(Class.class, Integer.TYPE);
        if (!constructor.isAccessible()) {
            constructor.setAccessible(true);
        }

        Class<?> declaringClass = method.getDeclaringClass();
        return (constructor.newInstance(declaringClass, 15)).unreflectSpecial(method, declaringClass).bindTo(proxy).invokeWithArguments(args);
    }
    private boolean isDefaultMethod(Method method) {
        return (method.getModifiers() & 1033) == 1 && method.getDeclaringClass().isInterface();
    }

    public Class<T> getiMapper() {
        return iMapper;
    }
}
